package com.lmm.rpc.core.client;

import com.lmm.rpc.core.client.protocol.MessageProtocol;
import com.lmm.rpc.core.client.protocol.RequestMetadata;
import com.lmm.rpc.core.cluster.RemoteServerInfo;
import com.lmm.rpc.core.codec.DefaultDecoder;
import com.lmm.rpc.core.codec.DefaultEncoder;
import com.lmm.rpc.core.common.reponse.DefaultRepose;
import com.lmm.rpc.core.common.request.DefaultRequest;
import com.lmm.rpc.core.exception.RpcException;
import com.lmm.rpc.core.handler.DefaultResponseHandler;
import com.lmm.rpc.core.transport.LocalRpcResponseCache;
import com.lmm.rpc.core.transport.RpcFuture;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.TimeUnit;


/**
 * netty 客户端
 * */
@Slf4j
@Data
public class NettyClient implements  NettyClientTransport<Channel>{
    private static final NioEventLoopGroup nioEventLoopGroup = new NioEventLoopGroup();
    private final Bootstrap bootstrap;
    private final DefaultResponseHandler responseHandler = new DefaultResponseHandler();

    public Bootstrap getBootstrap() {
        return bootstrap;
    }

    NettyClient(){
        this.bootstrap = new Bootstrap();
        this.open();
    }

    //初始化客户端
    private boolean open(){
        //初始化客户端    设置bootstrap的基本信息
        bootstrap.group(nioEventLoopGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        //业务处理器
                        socketChannel.pipeline()
                                //解码
                                .addLast(new DefaultDecoder())
                                .addLast(responseHandler)
                                //编码
                                .addLast(new DefaultEncoder<>());
                    }
                });
        return true;
    }

    /**
     * 获取连接好的channel
     *
     * @param remoteServer*/
    public Channel getConnectChannel(RemoteServerInfo remoteServer){
        return connect(remoteServer);
    }

    private Channel connect(RemoteServerInfo remoteServer){
        //避免网络抖动的原因 尝试3次
        int tryTime = 3;
        Channel channel = null;
        while (tryTime-->0){

            try {
                //同步建立连接
                ChannelFuture channelFuture = bootstrap.connect(remoteServer.getServer(), Integer.parseInt(remoteServer.getPort())).sync();
                log.info("connect rpc server {} on port {} success.",remoteServer.getServer(), remoteServer.getPort());
                channel = channelFuture.channel();
                return channel;
            } catch (Exception e) {
                log.warn("connect rpc server {} on port {} faild. retry {} again",remoteServer.getServer(), remoteServer.getPort(),3-tryTime);
                e.printStackTrace();
            }
        }
        throw new RpcException("connect rpc server on port faild finally"+remoteServer.getServer()+":"+remoteServer.getPort());
    }

    @Override
    public MessageProtocol<DefaultRepose> sendRequest(RequestMetadata metadata,Channel channel) throws Exception {
        MessageProtocol<DefaultRequest> protocol = metadata.getProtocol();
        RpcFuture<MessageProtocol<DefaultRepose>> reposeFuture = new RpcFuture<>();
        LocalRpcResponseCache.add(String.valueOf(protocol.getHeader().getRequestId()),reposeFuture);
        channel.writeAndFlush(protocol);
        log.info("send message success requestId:{}",protocol.getHeader().getRequestId());
        return metadata.getTimeout() != null ? reposeFuture.get(metadata.getTimeout(), TimeUnit.MILLISECONDS) : reposeFuture.get();
    }


//    @Override
//    public MessageProtocol<DefaultRepose> sendRequest(RequestMetadata metadata) throws Exception {
//        return null;
//    }
}
